package com.navinfo.opentsp.platform.computing.interceptor;

import com.google.common.base.Charsets;
import com.navinfo.opentsp.platform.computing.util.GsonUtil;
import com.navinfo.opentsp.platform.location.protocol.common.LCLocationData;
import org.apache.commons.lang.time.FastDateFormat;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.stream.Collectors;


/**
 * 改写Event中的header和body
 * <code>
 * agent.sources.r1.channels = c1<p>
 * agent.sources.r1.type = SEQ<p>
 * agent.sources.r1.interceptors = i1<p>
 * agent.sources.r1.interceptors.i1.type =  com.navinfo.opentsp.platform.computing.interceptor.HDFSLocationDataInterceptor$Builder<p>
 * agent.sources.r1.interceptors.i1.header = filePath<p>
 * agent.sources.r1.interceptors.i1.preserveExisting = true<p>
 * agent.sources.r1.interceptors.i1.datePattern= yyyy-MM-dd<p>
 * </code>
 */
public class HDFSLocationDataInterceptor implements Interceptor {
    private static final Logger logger = LoggerFactory.getLogger(HDFSLocationDataInterceptor.class);

    private String header;
    private boolean preserveExisting;
    private final FastDateFormat formatter;
    private final String splitStr = "\001";

    private HDFSLocationDataInterceptor(boolean preserveExisting, String header, String datePattern) {
        this.preserveExisting = preserveExisting;
        this.header = header;
        formatter = FastDateFormat.getInstance(datePattern);
    }

    @Override
    public Event intercept(Event event) {
        long time1 = System.currentTimeMillis();
        try {
            //拼写落盘到HDFS的body字符串。字段间分隔符是\001，null通过\N表示
            LCLocationData.LocationData locationData = LCLocationData.LocationData.parseFrom(event.getBody());

            StringBuilder locationStr = new StringBuilder(5000);
            locationStr
                .append(locationData.getAlarm()).append(splitStr)
                .append(locationData.getStatus()).append(splitStr)
                .append(locationData.getLongitude()).append(splitStr)
                .append(locationData.getLatitude()).append(splitStr)
                .append(locationData.getOriginalLng()).append(splitStr)
                .append(locationData.getOriginalLat()).append(splitStr)
                .append(locationData.getHeight()).append(splitStr)
                .append(locationData.getSpeed()).append(splitStr)
                .append(locationData.getDirection()).append(splitStr)
                .append(locationData.getGpsDate()).append(splitStr)
                .append(locationData.getMileage()).append(splitStr)
                .append(locationData.getReceiveDate()).append(splitStr)
                .append(locationData.getIsPatch()).append(splitStr)
                .append(locationData.getOil()).append(splitStr)
                .append(locationData.getRecorderSpeed()).append(splitStr)
                .append(locationData.getStarStatus()).append(splitStr)
                .append(locationData.getStarNumber()).append(splitStr)
                .append(locationData.getSignalStatus()).append(splitStr)
                .append(locationData.getIoStatus()).append(splitStr)
                .append(locationData.getAnalogAD0()).append(splitStr)
                .append(locationData.getAnalogAD1()).append(splitStr)
                .append(locationData.getSignalStrength()).append(splitStr)
                .append(locationData.getSatelliteNumber()).append(splitStr)
                .append(locationData.getIsValid()).append(splitStr)
                .append(GsonUtil.gsonString(locationData.getAdditionAlarm())).append(splitStr)
                .append(locationData.getBatteryPower()).append(splitStr)
                .append(locationData.getElectricVehicle()).append(splitStr)
                .append(locationData.getAlarmFilter()).append(splitStr)
                .append(locationData.getStandardMileage()).append(splitStr)
                .append(locationData.getStandardFuelCon()).append(splitStr)
                .append(locationData.getEcuDate()).append(splitStr)

                // 对应pb中子类型展开存
                .append(locationData.getTemperatureAddition().getVehType()).append(splitStr)
                .append(locationData.getTemperatureAddition().getTyrePosition()).append(splitStr)
                .append(locationData.getTemperatureAddition().getTyreCondition()).append(splitStr)
                .append(locationData.getTemperatureAddition().getTyrePressure()).append(splitStr)
                .append(locationData.getTemperatureAddition().getTyreTemperature()).append(splitStr)
                .append(locationData.getTemperatureAddition().getTemAlarmThreshold()).append(splitStr)
                .append(locationData.getTemperatureAddition().getTyreHalarmThreshold()).append(splitStr)
                .append(locationData.getTemperatureAddition().getTyreLalarmThreshold()).append(splitStr)
                .append(locationData.getTemperatureAddition().getTyreNomimalValue()).append(splitStr)

                .append(locationData.getAnalysisData().getTurningAngle()).append(splitStr)
                .append(locationData.getAnalysisData().getLowOilDrivingValue()).append(splitStr)
                .append(locationData.getAnalysisData().getVehicleSpeedFromEcu()).append(splitStr)
                .append(locationData.getAnalysisData().getRpmWhenAlarming()).append(splitStr)
                .append(locationData.getAnalysisData().getVelocityChangeValue()).append(splitStr)
                .append(locationData.getAnalysisData().getCurrentGearshift()).append(splitStr)
                .append(locationData.getAnalysisData().getCurrentRPM()).append(splitStr)
                .append(locationData.getAnalysisData().getBrakeTimes()).append(splitStr)
                .append(locationData.getAnalysisData().getClutchTimes()).append(splitStr)
                .append(locationData.getAnalysisData().getRetarderTimes()).append(splitStr)
                .append(locationData.getAnalysisData().getAbsTimes()).append(splitStr)
                .append(locationData.getAnalysisData().getReverseTimes()).append(splitStr)
                .append(locationData.getAnalysisData().getDeviceIdentity()).append(splitStr)
                .append(locationData.getBatteryInfo().getMotorTemperature()).append(splitStr)
                .append(locationData.getBatteryInfo().getMotorControlTemp()).append(splitStr)
                .append(locationData.getBatteryInfo().getMcuFaultCode()).append(splitStr)
                .append(locationData.getBatteryInfo().getBmsStatus()).append(splitStr)
                .append(locationData.getBatteryInfo().getBatteryAveTem()).append(splitStr)
                .append(locationData.getBatteryInfo().getBatteryCurrent()).append(splitStr)
                .append(locationData.getBatteryInfo().getTotalBatteryV()).append(splitStr)
                .append(locationData.getBatteryInfo().getBatteryFaultCode()).append(splitStr)
                .append(locationData.getBatteryInfo().getBatteryMaxTem()).append(splitStr)
                .append(locationData.getBatteryInfo().getBatteryMinTem()).append(splitStr)
                .append(locationData.getBatteryInfo().getBatteryMinSoc()).append(splitStr)
                .append(locationData.getBatteryInfo().getMaxChargingV()).append(splitStr)
                .append(locationData.getBatteryInfo().getMaxChargingE()).append(splitStr)
                .append(locationData.getBatteryInfo().getTotalStatus()).append(splitStr)
                .append(locationData.getBatteryInfo().getVehicleSpeed()).append(splitStr)
                .append(locationData.getBatteryInfo().getShiftStatus()).append(splitStr)
                .append(locationData.getBatteryInfo().getVcuFaultCode()).append(splitStr)
                .append(locationData.getBatteryInfo().getEnduranceMileage()).append(splitStr)
                .append(locationData.getBatteryInfo().getAcceleratorPedal()).append(splitStr)
                .append(locationData.getBatteryInfo().getBrakePedal()).append(splitStr)
                .append(locationData.getBatteryInfo().getModelInfo()).append(splitStr)
                .append(locationData.getBatteryInfo().getSwitchInfo()).append(splitStr)
                .append(locationData.getBatteryInfo().getElectricAttachment()).append(splitStr)
                .append(locationData.getBatteryInfo().getMotorRpm()).append(splitStr)

                // 对应pb中repeat的数据以json方式存入
                .append(GsonUtil.gsonString(locationData.getTemAlarmList())).append(splitStr)
                .append(GsonUtil.gsonString(locationData.getAlarmIdentifyList())).append(splitStr)
                .append(GsonUtil.gsonString(locationData.getDefenceAdditionList())).append(splitStr)
                .append(GsonUtil.gsonString(locationData.getParkingAdditionList())).append(splitStr)
                .append(GsonUtil.gsonString(locationData.getModuleVoltagesList())).append(splitStr)
                .append(GsonUtil.gsonString(locationData.getStaytimeParkingAdditionList())).append(splitStr)
                .append(GsonUtil.gsonString(locationData.getSpeedAdditionList())).append(splitStr)
                .append(GsonUtil.gsonString(locationData.getAreaAdditionList())).append(splitStr)
                .append(GsonUtil.gsonString(locationData.getRouteAdditionList())).append(splitStr)
                .append(GsonUtil.gsonString(locationData.getCancleBreakdownAddition().getBreakdownList())).append(splitStr)
                .append(GsonUtil.gsonString(locationData.getBreakdownAddition().getBreakdownList())).append(splitStr)
                .append(GsonUtil.gsonString(locationData.getStatusAddition()));

            //落盘到HDFS的目录结构topic/gpstime/commandId
            String filePath = "postrow/" + formatter.format(locationData.getGpsDate()) + "/" + locationData.getTerminalId();
            event.getHeaders().put(header, filePath);

            //将0200报文拼接后重新放入event的body中
            event.setBody(locationStr.toString().getBytes(Charsets.UTF_8));
        }catch (Exception e){
            logger.error("执行LocationData拦截器失败：" + e.getMessage(), e);
            return null;
        }finally {
            logger.info("执行LocationData拦截器总毫秒时间" + (System.currentTimeMillis() - time1));
        }
        return event;
    }

    @Override
    public List<Event> intercept(List<Event> events) {
        return events.stream().filter(event -> this.intercept(event)!=null).collect(Collectors.toList());
    }

    @Override
    public void initialize() { }

    @Override
    public void close() { }

    public static class Constants {
        public static final String PRESERVE = "preserveExisting";
        public static final boolean PRESERVE_DEF = true;
        public static final String HEADER = "header";
        public static final String HEADER_DEF = "filePath";
        public static final String DATEPATTERN = "datePattern";
        public static final String DATEPATTERN_DEF = "yyyy-MM-dd";

        public Constants() { }
    }

    public static class Builder implements Interceptor.Builder {
        private boolean preserveExisting;
        private String header;
        private String datePattern;
        public Builder() { }

        public void configure(Context context) {
            this.preserveExisting = context.getBoolean(Constants.PRESERVE, Constants.PRESERVE_DEF);
            this.header = context.getString(Constants.HEADER, Constants.HEADER_DEF);
            this.datePattern = context.getString(Constants.DATEPATTERN, Constants.DATEPATTERN_DEF);
        }

        public Interceptor build() {
            return new HDFSLocationDataInterceptor(this.preserveExisting, this.header, this.datePattern);
        }
    }

}
